package org.fun.kafkademo.consumer;

import org.fun.kafkademo.model.BatchProcessResult;
import org.fun.kafkademo.model.UserEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 批量数据消费者
 */
@Component
public class BatchConsumer {
    
    private static final Logger logger = LoggerFactory.getLogger(BatchConsumer.class);
    
    // 统计信息
    private final AtomicLong totalProcessed = new AtomicLong(0);
    private final AtomicLong totalSuccess = new AtomicLong(0);
    private final AtomicLong totalFailure = new AtomicLong(0);
    private final Map<String, Long> eventTypeCounts = new ConcurrentHashMap<>();
    
    /**
     * 批量消费用户事件数据
     */
    @KafkaListener(topics = "user-events", groupId = "kafka-batch-consumer-group", 
                   containerFactory = "kafkaListenerContainerFactory")
    public void handleBatchUserEvents(@Payload List<UserEvent> events,
                                    @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                    @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                                    @Header(KafkaHeaders.OFFSET) List<Long> offsets,
                                    Acknowledgment acknowledgment) {
        
        LocalDateTime startTime = LocalDateTime.now();
        int successCount = 0;
        int failureCount = 0;
        List<String> errors = new ArrayList<>();
        
        logger.info("开始批量处理 {} 条用户事件数据", events.size());
        
        try {
            for (UserEvent event : events) {
                try {
                    // 处理单个用户事件
                    processUserEvent(event);
                    successCount++;
                    
                    // 更新统计信息
                    eventTypeCounts.merge(event.getEventType(), 1L, Long::sum);
                    
                } catch (Exception e) {
                    failureCount++;
                    String error = "处理事件失败: " + event.getUserId() + " - " + e.getMessage();
                    errors.add(error);
                    logger.error("处理用户事件失败: {}", event, e);
                }
            }
            
            // 更新全局统计
            totalProcessed.addAndGet(events.size());
            totalSuccess.addAndGet(successCount);
            totalFailure.addAndGet(failureCount);
            
            LocalDateTime endTime = LocalDateTime.now();
            long processingTime = java.time.Duration.between(startTime, endTime).toMillis();
            
            logger.info("批量处理完成: 总数={}, 成功={}, 失败={}, 耗时={}ms", 
                events.size(), successCount, failureCount, processingTime);
            
            // 手动提交偏移量
            acknowledgment.acknowledge();
            
        } catch (Exception e) {
            logger.error("批量处理用户事件失败", e);
            // 可以选择是否提交偏移量，这里选择不提交以便重试
            // acknowledgment.acknowledge();
        }
    }
    
    /**
     * 处理单个用户事件
     */
    private void processUserEvent(UserEvent event) {
        // 模拟业务处理逻辑
        logger.debug("处理用户事件: userId={}, eventType={}, timestamp={}", 
            event.getUserId(), event.getEventType(), event.getTimestamp());
        
        // 模拟不同的处理逻辑
        switch (event.getEventType()) {
            case "LOGIN":
                processLoginEvent(event);
                break;
            case "LOGOUT":
                processLogoutEvent(event);
                break;
            case "PAGE_VIEW":
                processPageViewEvent(event);
                break;
            case "CLICK":
                processClickEvent(event);
                break;
            case "PURCHASE":
                processPurchaseEvent(event);
                break;
            case "SEARCH":
                processSearchEvent(event);
                break;
            default:
                logger.warn("未知的事件类型: {}", event.getEventType());
        }
        
        // 模拟处理时间
        try {
            Thread.sleep(10 + (long)(Math.random() * 20)); // 10-30ms
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
    
    private void processLoginEvent(UserEvent event) {
        logger.debug("处理登录事件: userId={}, sessionId={}", 
            event.getUserId(), event.getSessionId());
        // 实际业务逻辑：记录登录日志、更新用户状态等
    }
    
    private void processLogoutEvent(UserEvent event) {
        logger.debug("处理登出事件: userId={}, sessionId={}", 
            event.getUserId(), event.getSessionId());
        // 实际业务逻辑：清理会话、记录登出时间等
    }
    
    private void processPageViewEvent(UserEvent event) {
        logger.debug("处理页面浏览事件: userId={}, data={}", 
            event.getUserId(), event.getEventData());
        // 实际业务逻辑：记录页面访问、更新用户行为等
    }
    
    private void processClickEvent(UserEvent event) {
        logger.debug("处理点击事件: userId={}, data={}", 
            event.getUserId(), event.getEventData());
        // 实际业务逻辑：记录点击行为、分析用户兴趣等
    }
    
    private void processPurchaseEvent(UserEvent event) {
        logger.debug("处理购买事件: userId={}, data={}", 
            event.getUserId(), event.getEventData());
        // 实际业务逻辑：处理订单、更新用户购买记录等
    }
    
    private void processSearchEvent(UserEvent event) {
        logger.debug("处理搜索事件: userId={}, data={}", 
            event.getUserId(), event.getEventData());
        // 实际业务逻辑：记录搜索关键词、分析搜索行为等
    }
    
    /**
     * 获取处理统计信息
     */
    public BatchProcessResult getProcessingStats() {
        return new BatchProcessResult(
            totalProcessed.intValue(),
            totalSuccess.intValue(),
            totalFailure.intValue(),
            new ArrayList<>(),
            LocalDateTime.now().minusHours(1), // 假设开始时间
            LocalDateTime.now()
        );
    }
    
    /**
     * 获取事件类型统计
     */
    public Map<String, Long> getEventTypeCounts() {
        return new ConcurrentHashMap<>(eventTypeCounts);
    }
    
    /**
     * 重置统计信息
     */
    public void resetStats() {
        totalProcessed.set(0);
        totalSuccess.set(0);
        totalFailure.set(0);
        eventTypeCounts.clear();
        logger.info("统计信息已重置");
    }
}
